package com.mikasa.spring;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import com.mikasa.spring.adapter.MessageDelegate;
import com.mikasa.spring.convert.ImageMessageConverter;
import com.mikasa.spring.convert.PDFMessageConverter;
import com.mikasa.spring.convert.TextMessageConverter;

/**
 * 配置类
 * @author mikasa
 */
@Configuration
@ComponentScan({"com.mikasa.spring.*"})
public class RabbitMQConfig {

	/**
	 * 配置连接信息
	 * @return
	 */
	@Bean
	public ConnectionFactory connectionFactory(){
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		// ip+port
		connectionFactory.setAddresses("39.108.108.234:5672");
		connectionFactory.setUsername("guest");
		connectionFactory.setPassword("guest");
		connectionFactory.setVirtualHost("/");
		return connectionFactory;
	}

	/**
	 * 创建Admin管理对象用于操作mq如声明交换机、队列等等（API示例请看测试类）
	 *
	 * @param connectionFactory
	 * @return
	 */
	@Bean
	public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
		// 初始化，必须为true，在springbean加载后进行初始化，将所有类型为 exchange、queue、binging对象全部加载进指定的集合中
		rabbitAdmin.setAutoStartup(true);
		return rabbitAdmin;
	}
	
    /**  
     * 针对消费者配置  
     * 1. 设置交换机类型  
     * 2. 将队列绑定到交换机  
        FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念  
        HeadersExchange ：通过添加属性key-value匹配  
        DirectExchange:按照routingkey分发到指定队列  
        TopicExchange:多关键字匹配  
     */

	/**
	 * SpringAMQP中的交换机的声明
	 * @return
	 */
	@Bean
    public TopicExchange exchange001() {  
        return new TopicExchange("topic001", true, false);  
    }

	/**
	 *  SpringAMQP中的队列的声明
	 * @return
	 */
	@Bean
    public Queue queue001() {  
        return new Queue("queue001", true);
    }

	/**
	 *  SpringAMQP中绑定的声明
	 * @return
	 */
	@Bean
    public Binding binding001() {  
        return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");  
    }

	@Bean
	public TopicExchange exchange002() {
		return new TopicExchange("topic002", true, false);
	}

	@Bean
	public Queue queue002() {
		return new Queue("queue002", true);
	}

	@Bean
	public Binding binding002() {
		return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");
	}

	@Bean
	public Queue queue003() {
		return new Queue("queue003", true);
	}

	@Bean
	public Binding binding003() {
		return BindingBuilder.bind(queue003()).to(exchange001()).with("mq.*");
	}

	@Bean
	public Queue queue_image() {
		return new Queue("image_queue", true);
	}

	@Bean
	public Queue queue_pdf() {
		return new Queue("pdf_queue", true);
	}

	/**
	 * 消息模板，SpringAMQP整合进行发送消息的关键类 （springboot中只需要在配置文件中配置信息即可）
	 * API示例见 test类
	 * @param connectionFactory
	 * @return
	 */
	@Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    	return rabbitTemplate;
    }

	/**
	 * 简单消息监听容器
	 * 		-	可以对消费者进行丰富的配置
	 *		-	监听多个队列...
	 *		-	设置事务特性、事务管理器、事务属性...
	 *		-	设置消费者数量、最大最小数量、批量消费...
	 *		-	设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数等
	 *		-	设置消费者标签生产策略、是否独占模式、消费者属性等
	 *		-	设置具体监听器、转换器
	 *		-	在运行中可以动态的修改消费者数量的大小、接收消息的模式等
	 *
	 * @param connectionFactory
	 * @return
	 */
	@Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
    	
    	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    	// 监听队列
		container.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
		// 当前消费者数量
    	container.setConcurrentConsumers(1);
		// 最大消费者数量
    	container.setMaxConcurrentConsumers(5);
		// 是否
    	container.setDefaultRequeueRejected(false);
		// 签收模式
    	container.setAcknowledgeMode(AcknowledgeMode.AUTO);
    	container.setExposeListenerChannel(true);
		// 消费端标签策略
    	container.setConsumerTagStrategy(new ConsumerTagStrategy() {
			@Override
			public String createConsumerTag(String queue) {
				return queue + "_" + UUID.randomUUID().toString();
			}
		});


		/**
		 * MessageListener 消息监听
		 */
    	container.setMessageListener(new ChannelAwareMessageListener() {
			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				String msg = new String(message.getBody());
				System.err.println("----------消费者: " + msg);
			}
		});

		/**
		 * MessageListenerAdapter 消息监听适配器
		 *		 MessageListenerAdapter类中
		 *		 public static final String ORIGINAL_DEFAULT_LISTENER_METHOD = "handleMessage";
		 * 		 规定了自定义适配器类必须要有 handleMessage方法，反射根据这个名字把监听的消息传进去
		 */

		/**
		 * 	适配器方式1：
		 */
//		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//		// 转换器，将接收到的消息进行自定义转化
//		// 直接接收message消息内容是 byte[]数组，而自定义适配器中监听方法 consumeMessage(String messageBody) 是字符串类型的
//		// 此时直接接收会有类型错误，需要进行转换
//		adapter.setMessageConverter(new TextMessageConverter());
//		// 也可自手动设置默认方法
//    	adapter.setDefaultListenerMethod("consumeMessage");
//    	container.setMessageListener(adapter);

		/**
		 * 	适配器方式2: 队列名称 和 方法名称 也可以进行一一的匹配
		 */
//    	MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//    	adapter.setMessageConverter(new TextMessageConverter());
//		// 将队列和需要路由过去的方法绑定起来
//    	Map<String, String> queueOrTagToMethodName = new HashMap<>();
//		// 监听的queue001队列监听到消息就会访问 method1 这个方法
//    	queueOrTagToMethodName.put("queue001", "method1");
//    	queueOrTagToMethodName.put("queue002", "method2");
//    	adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
//    	container.setMessageListener(adapter);

		/**
		 * MessageConverter 消息转换器
		 * 	正常情况下消息体为二进制的数据方式进行传输，如果希望内部进行转换，或者指定自定义的转换器，就需要用到MessageConverter
		 * 	自定义转换器需要实现 MessageConverter接口
		 * 		重写 toMessage() fromMessage()两个方法
		 * 		常见有：	 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象转换、java对象多映射转换
		 * 				 自定义的 二进制转换 文本、图片、pdf、流媒体等转换
		 */

		/**
		 * json格式的转换器
		 *	消息发送需设置 messageProperties.setContentType 为 application/json
		 *		示例见 test类的 testSendJsonMessage() 方法
		 * 	自定义消息监听器接收消息的指定方法入参需为 map ,json消息对应的java对象就是map
		 */
//		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setDefaultListenerMethod("consumeMessage");
//        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//        adapter.setMessageConverter(jackson2JsonMessageConverter);
//        container.setMessageListener(adapter);

		/**
		 * DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象转换
		 * 	消息发送需设置 messageProperties.setContentType 为 application/json
		 * 		示例见 test类的 testSendJsonMessage() 方法
		 * 	自定义消息监听器接收消息的指定方法入参需为 发送消息时的对象
		 */
//		 MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setDefaultListenerMethod("consumeMessage");
//        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//
//        DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
//        jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
//
//        adapter.setMessageConverter(jackson2JsonMessageConverter);
//        container.setMessageListener(adapter);

		/**
		 * DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象多映射转换
		 * 	需要先声明 java类与标签的映射关系，发送消息时传入对应标签即可
		 *	消息发送示例见 test类的 testSendMappingMessage() 方法
		 */
//		 MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setDefaultListenerMethod("consumeMessage");
//        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//
//		// 创建一个映射类
//		DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
//
//        Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
//		// 将标签和对应的全限定名进行映射，发送消息时只需要在Headers的__TypeId__中put指定类对应的标签即可
//		idClassMapping.put("order", com.mikasa.spring.entity.Order.class);
//		idClassMapping.put("packaged", com.mikasa.spring.entity.Packaged.class);
//		// 设置映射
//		javaTypeMapper.setIdClassMapping(idClassMapping);
//		jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
//
//        adapter.setMessageConverter(jackson2JsonMessageConverter);
//        container.setMessageListener(adapter);

		/**
		 * 全局转换器：ContentTypeDelegatingMessageConverter
		 * 	大的转换器包含多个小转换器
		 */
		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");
        
        // 全局的转换器:
		ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();

		// 文本转换
		TextMessageConverter textConvert = new TextMessageConverter();
		convert.addDelegate("text", textConvert);
		convert.addDelegate("html/text", textConvert);
		convert.addDelegate("xml/text", textConvert);
		convert.addDelegate("text/plain", textConvert);

		// json转换
		Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();
		convert.addDelegate("json", jsonConvert);
		convert.addDelegate("application/json", jsonConvert);

		// 图片转换
		ImageMessageConverter imageConverter = new ImageMessageConverter();
		convert.addDelegate("image/png", imageConverter);
		convert.addDelegate("image", imageConverter);

		// pdf转换
		PDFMessageConverter pdfConverter = new PDFMessageConverter();
		convert.addDelegate("application/pdf", pdfConverter);

		// 视频转换...
        
		
		adapter.setMessageConverter(convert);
		container.setMessageListener(adapter);
		
    	return container;
    	
    }
    
    
    
    
    
    
    
    
    
    
    
	
	
	
}
